/******************************************************************************/
/* @ All Rights Reserved Copyright (C) 2014 broada,Ltd.                       */
/*                                                                            */
/*  NAME      = kafka_comm.c                                                  */
/*  NOTE      = KAFKA请求解析                                                 */
/*  DATE      = 2015/11/10 by zhengxd                                         */
/******************************************************************************/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <math.h>
#include <stdint.h>
#include <limits.h>

#include "kafka_core.h"
#include "kafka_comm.h"

/*******************************************************************************
*  FUNC     :  调试用
*  ARGS     :  void
*  RTN      :  void
*  NOTE     :
*******************************************************************************/
void kafka_dbg(void)
{
    return;
}

/*******************************************************************************
*  FUNC     :  释放内存
*  ARGS     :  pp_buffer
*  RTN      :  void
*  NOTE     :
*******************************************************************************/
void kafka_free(char **pp_buffer)
{
    if (pp_buffer == NULL || *pp_buffer == NULL)
    {
        return;
    }

    frfree(*pp_buffer);
    *pp_buffer = NULL;
    return;
}

/*******************************************************************************
*  FUNC     :  取得short值
*  ARGS     :  p_st_data(入力buff)
*           :  p_i_len(出力长度)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_TODO
*              APM_L7_ANALYZE_FAIL
*  NOTE     :
*******************************************************************************/
int kafka_get_short(Frbyte_Packet *p_st_data, int *p_i_len)
{
    int i_ret = FR_SUCC;
    short s_len = 0;
    uint16_t u16_len = 0;

    /* get length */
    i_ret = bytebuff_get_short(p_st_data, 2, &u16_len, BYTE_BIG_ENDIAN);
    if (i_ret != FR_SUCC)
    {
        return APM_L7_ANALYZE_FAIL;
    }

    s_len = (short)u16_len;
    *p_i_len = s_len;
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  取得int值
*  ARGS     :  p_st_data(入力buff)
*           :  p_i_len(出力长度)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_TODO
*              APM_L7_ANALYZE_FAIL
*  NOTE     :
*******************************************************************************/
int kafka_get_int(Frbyte_Packet *p_st_data, int *p_i_len)
{
    int i_ret = FR_SUCC;
    uint32_t u32_len = 0;

    /* get length */
    i_ret = bytebuff_get_int(p_st_data, 4, &u32_len, BYTE_BIG_ENDIAN);
    if (i_ret != FR_SUCC)
    {
        return APM_L7_ANALYZE_FAIL;
    }

    *p_i_len = (int)u32_len;
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  取得string
*  ARGS     :  p_st_data(入力buff)
*           :  pp_out_str(出力string)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_TODO
*              APM_L7_ANALYZE_FAIL
*  NOTE     :
*******************************************************************************/
int kafka_get_string(Frbyte_Packet *p_st_data, char **pp_out_str)
{
    int i_ret = FR_SUCC;
    uint16_t u16_str_len = 0;
    char *p_out = NULL;

    /* get length */
    i_ret = bytebuff_get_short(p_st_data, 2, &u16_str_len, BYTE_BIG_ENDIAN);
    if (i_ret != FR_SUCC)
    {
        return APM_L7_ANALYZE_FAIL;
    }

    /* 如果长度是-1,则内容为NULL */
    if (u16_str_len == 0xFFFF || u16_str_len == 0)
    {
        *pp_out_str = NULL;
        return APM_L7_ANALYZE_SUCC;
    }

    if (u16_str_len > bytebuff_remaining(p_st_data))
    {
        return APM_L7_ANALYZE_FAIL;
    }
    p_out = (char *)frmalloc(u16_str_len + 1);
    if (p_out == NULL)
    {
        return APM_L7_ANALYZE_FAIL;
    }
    i_ret = bytebuff_get_bytes(p_st_data, p_out, u16_str_len);
    if (i_ret != FR_SUCC)
    {
        frfree(p_out);
        return APM_L7_ANALYZE_FAIL;
    }
    p_out[u16_str_len] = '\0';

    *pp_out_str = p_out;
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  取得bytes
*  ARGS     :  p_st_data(入力buff)
*           :  pp_out_bytes(出力bytes)
*           :  p_ui_size(出力bytes的size)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_TDO
*              APM_L7_ANALYZE_FAIL
*  NOTE     :
*******************************************************************************/
int kafka_get_bytes(Frbyte_Packet *p_st_data
                    , int i_attributes
                    , char **pp_out_bytes
                    , unsigned int *p_ui_size)
{
    int i_ret = FR_SUCC;
    uint32_t u32_byte_len = 0;
    char *p_out = NULL;

    /* get length */
    i_ret = bytebuff_get_int(p_st_data, 4, &u32_byte_len, BYTE_BIG_ENDIAN);
    if (i_ret != FR_SUCC)
    {
        return APM_L7_ANALYZE_FAIL;
    }

    /* 如果长度是-1,则内容为NULL */
    if (u32_byte_len == 0xFFFFFFFF || u32_byte_len == 0)
    {
        *pp_out_bytes = NULL;
        return APM_L7_ANALYZE_SUCC;
    }

    if (u32_byte_len > (unsigned int)bytebuff_remaining(p_st_data))
    {
        return APM_L7_ANALYZE_FAIL;
    }
    p_out = (char *)frmalloc(u32_byte_len + 1);
    if (p_out == NULL)
    {
        return APM_L7_ANALYZE_FAIL;
    }
    i_ret = bytebuff_get_bytes(p_st_data, p_out, u32_byte_len);
    if (i_ret != FR_SUCC)
    {
        frfree(p_out);
        return APM_L7_ANALYZE_FAIL;
    }
    p_out[u32_byte_len] = '\0';

    //todo 根据i_attributes解压缩
    *pp_out_bytes = p_out;
    *p_ui_size = u32_byte_len;
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  跳过string
*  ARGS     :  p_st_data(入力buff)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_FAIL
*  NOTE     :
*******************************************************************************/
int kafka_skip_string(Frbyte_Packet *p_st_data)
{
    int i_ret = FR_SUCC;
    uint16_t u16_str_len = 0;

    /* get length */
    i_ret = bytebuff_get_short(p_st_data, 2, &u16_str_len, BYTE_BIG_ENDIAN);
    if (i_ret != FR_SUCC)
    {
        return APM_L7_ANALYZE_FAIL;
    }

    /* 如果长度是-1,则内容为NULL */
    if (u16_str_len == 0xFFFF)
    {
        return APM_L7_ANALYZE_SUCC;
    }

    /* skip string */
    i_ret = bytebuff_skip(p_st_data, u16_str_len);
    if (i_ret != FR_SUCC)
    {
        return APM_L7_ANALYZE_FAIL;
    }
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  跳过bytes
*  ARGS     :  p_st_data(入力buff)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_TDO
*              APM_L7_ANALYZE_FAIL
*  NOTE     :
*******************************************************************************/
int kafka_skip_bytes(Frbyte_Packet *p_st_data)
{
    int i_ret = FR_SUCC;
    uint32_t u32_byte_len = 0;

    /* get length */
    i_ret = bytebuff_get_int(p_st_data, 4, &u32_byte_len, BYTE_BIG_ENDIAN);
    if (i_ret != FR_SUCC)
    {
        return APM_L7_ANALYZE_FAIL;
    }

    /* 如果长度是-1,则内容为NULL */
    if (u32_byte_len == 0xFFFFFFFF)
    {
        return APM_L7_ANALYZE_SUCC;
    }

    i_ret = bytebuff_skip(p_st_data, u32_byte_len);
    if (i_ret != FR_SUCC)
    {
        return APM_L7_ANALYZE_FAIL;
    }

    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  解析message set
*  ARGS     :  p_st_data(入力buff)
*           :  ui_message_set_size(message set的size)
*           :  p_st_message_set(解析生成的message)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_FAIL
*  NOTE     :  p_st_message_set必须不是NULL
*******************************************************************************/
int kafka_parse_message_set(Frbyte_Packet *p_st_data
                           , unsigned int ui_message_set_size
                           , Frlist_header *p_st_message_set)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    unsigned int ui_before_len = 0;
    uint8_t u8_attributes = 0;
    unsigned int ui_value_len = 0;
    char *p_value = NULL;
    Kafka_message_set_info *p_st_message = NULL;
    Frlist_node *p_st_node = NULL;

    /* 解析message set */
    while (ui_message_set_size)
    {
        ui_before_len = bytebuff_remaining(p_st_data);

        /* skip offset and message size and Crc and MagicByte */
        i_ret = bytebuff_skip(p_st_data, 8 + 4 + 4 + 1);
        if (i_ret != FR_SUCC)
        {
            frdbg("skip bytes failed!\n");
            frlist_remove_all(p_st_message_set, kafka_message_set_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        /* get attributes */
        i_ret = bytebuff_get_char(p_st_data, &u8_attributes);
        if (i_ret != FR_SUCC)
        {
            frdbg("get attributes failed!\n");
            frlist_remove_all(p_st_message_set, kafka_message_set_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        else if (u8_attributes > 2)
        {
            frdbg("attributes is incorrect!\n");
            frlist_remove_all(p_st_message_set, kafka_message_set_node_free);
            return APM_L7_ANALYZE_FAIL;
        }

        /* skip Key */
        i_ret = kafka_skip_bytes(p_st_data);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            frdbg("skip key failed!\n");
            frlist_remove_all(p_st_message_set, kafka_message_set_node_free);
            return i_ret;
        }
        /* get value */
        i_ret = kafka_get_bytes(p_st_data, u8_attributes, &p_value, &ui_value_len);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            frdbg("get value failed!\n");
            frlist_remove_all(p_st_message_set, kafka_message_set_node_free);
            return i_ret;
        }

        /* 创建message set */
        p_st_message = (Kafka_message_set_info *)frmalloc(sizeof(Kafka_message_set_info));
        if (p_st_message == NULL)
        {
            kafka_free(&p_value);
            frlist_remove_all(p_st_message_set, kafka_message_set_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        p_st_message->p_value = p_value;

        /* 格式不对 */
        if (ui_message_set_size < ui_before_len - bytebuff_remaining(p_st_data))
        {
            kafka_message_set_free(&p_st_message);
            frlist_remove_all(p_st_message_set, kafka_message_set_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        ui_message_set_size -= ui_before_len - bytebuff_remaining(p_st_data);

        /* 创建节点 */
        p_st_node = frlist_node_create(p_st_message);
        if (p_st_node == NULL)
        {
            kafka_message_set_free(&p_st_message);
            frlist_remove_all(p_st_message_set, kafka_message_set_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        /* 追加节点 */
        frlist_push(p_st_message_set, p_st_node);
    }

    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  Kafka_message_set_info节点释放
*  ARGS     :  p_st_node(Kafka_message_set_info *)
*  RTN      :  void
*  NOTE     :
*******************************************************************************/
void kafka_message_set_node_free(Frlist_node *p_st_node)
{
    if( p_st_node == NULL || p_st_node->data == NULL)
    {
        return;
    }

    Kafka_message_set_info *p_st_message_set = (Kafka_message_set_info *)p_st_node->data;
    if (p_st_message_set->p_value)
    {
        frfree(p_st_message_set->p_value);
    }
    frfree(p_st_node->data);

    p_st_node->data = NULL;
    return;
}

/*******************************************************************************
*  FUNC     :  message set信息释放
*  ARGS     :  pp_st_message_set(message set信息)
*  RTN      :  void
*  NOTE     :
*******************************************************************************/
void kafka_message_set_free(Kafka_message_set_info **pp_st_message_set)
{
    if (pp_st_message_set == NULL || *pp_st_message_set == NULL)
    {
        return;
    }

    Kafka_message_set_info *p_st_message_set = *pp_st_message_set;
    if (p_st_message_set->p_value)
    {
        frfree(p_st_message_set->p_value);
    }
    frfree(p_st_message_set);

    *pp_st_message_set = NULL;
    return;
}

/*******************************************************************************
*  FUNC     :  Kafka_partition_info节点释放
*  ARGS     :  p_st_node(Kafka_partition_info *)
*  RTN      :  void
*  NOTE     :
*******************************************************************************/
void kafka_partition_node_free(Frlist_node *p_st_node)
{
    if( p_st_node == NULL || p_st_node->data == NULL)
    {
        return;
    }

    Kafka_partition_info *p_st_partition = (Kafka_partition_info *)p_st_node->data;
    frlist_remove_all(&p_st_partition->st_messages, kafka_message_set_node_free);
    frfree(p_st_node->data);

    p_st_node->data = NULL;
    return;
}

/*******************************************************************************
*  FUNC     :  partition信息释放
*  ARGS     :  pp_st_partition(partition信息)
*  RTN      :  void
*  NOTE     :
*******************************************************************************/
void kafka_partition_free(Kafka_partition_info **pp_st_partition)
{
    if (pp_st_partition == NULL || *pp_st_partition == NULL)
    {
        return;
    }

    Kafka_partition_info *p_st_partition = *pp_st_partition;
    frlist_remove_all(&p_st_partition->st_messages, kafka_message_set_node_free);
    frfree(p_st_partition);

    *pp_st_partition = NULL;
    return;
}

/*******************************************************************************
*  FUNC     :  topic信息释放
*  ARGS     :  pp_st_topic(topic信息)
*  RTN      :  void
*  NOTE     :
*******************************************************************************/
void kafka_topic_free(Kafka_topic_info **pp_st_topic)
{
    if (pp_st_topic == NULL || *pp_st_topic == NULL)
    {
        return;
    }

    Kafka_topic_info *p_st_topic = *pp_st_topic;
    if (p_st_topic->p_topic_name)
    {
        frfree(p_st_topic->p_topic_name);
    }
    frlist_remove_all(&p_st_topic->st_partitions, kafka_partition_node_free);
    frfree(p_st_topic);

    *pp_st_topic = NULL;
    return;
}

/*******************************************************************************
*  FUNC     :  Kafka_topic_info节点释放
*  ARGS     :  p_st_node(Kafka_topic_info *)
*  RTN      :  void
*  NOTE     :
*******************************************************************************/
void kafka_topic_node_free(Frlist_node *p_st_node)
{
    if( p_st_node == NULL || p_st_node->data == NULL)
    {
        return;
    }

    Kafka_topic_info *p_st_topic = (Kafka_topic_info *)p_st_node->data;
    if (p_st_topic->p_topic_name)
    {
        frfree(p_st_topic->p_topic_name);
    }
    frlist_remove_all(&p_st_topic->st_partitions, kafka_partition_node_free);
    frfree(p_st_node->data);

    p_st_node->data = NULL;
    return;
}

/*******************************************************************************
*  FUNC     :  request信息释放
*  ARGS     :  pp_st_req(request信息)
*  RTN      :  void
*  NOTE     :
*******************************************************************************/
void kafka_destroy_request(L7_req_kafka **pp_st_req)
{
    if (pp_st_req == NULL || *pp_st_req == NULL)
    {
        return;
    }

    L7_req_kafka *p_st_req = *pp_st_req;
    frlist_remove_all(&p_st_req->st_topics, kafka_topic_node_free);
    frfree(p_st_req);

    *pp_st_req = NULL;
    return;
}

/*******************************************************************************
*  FUNC     :  response信息释放
*  ARGS     :  pp_st_resp(response信息)
*  RTN      :  void
*  NOTE     :
*******************************************************************************/
void kafka_destroy_response(L7_resp_kafka **pp_st_resp)
{
    if (pp_st_resp == NULL || *pp_st_resp == NULL)
    {
        return;
    }

    L7_resp_kafka *p_st_resp = *pp_st_resp;
    frlist_remove_all(&p_st_resp->st_topics, kafka_topic_node_free);
    frfree(p_st_resp);

    *pp_st_resp = NULL;
    return;
}

/*******************************************************************************
*  FUNC     :  分片包初始化
*  ARGS     :  p_st_fragment(分片信息)
*  RTN      :  void
*  NOTE     :
*******************************************************************************/
void kafka_init_fragment(L7_fragment_kafka *p_st_fragment)
{
    if (p_st_fragment == NULL)
    {
        return;
    }

    if (p_st_fragment->p_st_data)
    {
        bytebuff_free(&p_st_fragment->p_st_data);
    }
    memset(p_st_fragment, 0x0, sizeof(L7_fragment_kafka));

    return;
}

/*******************************************************************************
*  FUNC     :  分片包释放
*  ARGS     :  pp_st_fragment(Kafka_topic_req_info *)
*  RTN      :  void
*  NOTE     :
*******************************************************************************/
void kafka_destroy_fragment(L7_fragment_kafka **pp_st_fragment)
{
    if (pp_st_fragment == NULL || *pp_st_fragment == NULL)
    {
        return;
    }

    L7_fragment_kafka *p_st_fragment = *pp_st_fragment;
    if (p_st_fragment->p_st_data)
    {
        bytebuff_free(&p_st_fragment->p_st_data);
    }
    frfree(p_st_fragment);

    *pp_st_fragment = NULL;
    return;
}

/*******************************************************************************
*  FUNC     :  取得时间差(微秒)
*  ARGS     :  st_start_time(开始时间)
*           :  p_st_end_time(结束时间)
*  RTN      :  long
*  NOTE     :
*******************************************************************************/
unsigned long kafka_get_interval_time(struct timeval *p_st_start_time
                                       , struct timeval *p_st_end_time)
{
    unsigned long ul_time_interval = 0;
    int i_flg = 0;

    if (p_st_start_time->tv_usec < p_st_end_time->tv_usec)
    {
        ul_time_interval = p_st_end_time->tv_usec - p_st_start_time->tv_usec;
    }
    else
    {
        ul_time_interval = 1000000 + p_st_end_time->tv_usec - p_st_start_time->tv_usec;
        i_flg = 1;
    }

    if (p_st_start_time->tv_sec + i_flg > p_st_end_time->tv_sec)
    {
        ul_time_interval = 0;
    }
    else
    {
        ul_time_interval += 1000000 * (p_st_end_time->tv_sec - p_st_start_time->tv_sec - i_flg);
    }

    return ul_time_interval;
}

/*******************************************************************************
*  FUNC     :  取得message version
*  ARGS     :  p_st_data(入力buff)
*           :  ui_size(message size)
*           :  p_i_msg_ver(出力message version)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_FAIL
*  NOTE     :
*******************************************************************************/
int kafka_get_msg_version(Frbyte_Packet *p_st_data
                          , unsigned int ui_size
                          , int *p_i_msg_ver)
{
    /*
    * 0.6的Message = LENGTH + MAGIC + CHECKSUM + PAYLOAD
    * 0.7的Message = LENGTH + MAGIC + COMPRESSION + CHECKSUM + PAYLOAD
    */
    int i_ret = FR_SUCC;
    int i_confirm_flg = FR_YES;
    uint8_t u8_magic = 0;
    uint8_t u8_compression = 0;
    uint32_t u32_msg_len = 0;
    uint32_t u32_payload_len = 0;
    unsigned int ui_position = bytebuff_getPosition(p_st_data);
    unsigned int ui_size_tmp = ui_size;

    /* 先判定是否是0.6的 */
    while (ui_size_tmp)
    {
        /* get length */
        i_ret = bytebuff_get_int(p_st_data, 4, &u32_msg_len, BYTE_BIG_ENDIAN);
        if (i_ret != FR_SUCC || ui_size < 4 + u32_msg_len)
        {
            i_confirm_flg = FR_NO;
            break;
        }
        i_ret = bytebuff_get_char(p_st_data, &u8_magic);
        if (i_ret != FR_SUCC || u8_magic != 0)
        {
            i_confirm_flg = FR_NO;
            break;
        }
        /* skip CHECKSUM */
        i_ret = bytebuff_skip(p_st_data, 4);
        if (i_ret != FR_SUCC)
        {
            i_confirm_flg = FR_NO;
            break;
        }
        /* get length */
        i_ret = bytebuff_get_int(p_st_data, 4, &u32_payload_len, BYTE_BIG_ENDIAN);
        if (i_ret != FR_SUCC || u32_msg_len != 1 + 4 + 4 + u32_payload_len)
        {
            i_confirm_flg = FR_NO;
            break;
        }
        i_ret = bytebuff_skip(p_st_data, u32_payload_len);
        if (i_ret != FR_SUCC)
        {
            i_confirm_flg = FR_NO;
            break;
        }
        ui_size_tmp -= u32_msg_len + 4;
    }
    bytebuff_setPosition(p_st_data, ui_position);
    /* 是0.6 */
    if (i_confirm_flg == FR_YES)
    {
        *p_i_msg_ver = KAFKA_VERSION_6;
        return APM_L7_ANALYZE_SUCC;
    }

    /* 再判定是否是0.7的 */
    ui_size_tmp = ui_size;
    while (ui_size_tmp)
    {
        /* get length */
        i_ret = bytebuff_get_int(p_st_data, 4, &u32_msg_len, BYTE_BIG_ENDIAN);
        if (i_ret != FR_SUCC || ui_size < 4 + u32_msg_len)
        {
            i_confirm_flg = FR_NO;
            break;
        }
        i_ret = bytebuff_get_char(p_st_data, &u8_magic);
        if (i_ret != FR_SUCC || u8_magic != 0)
        {
            i_confirm_flg = FR_NO;
            break;
        }
        i_ret = bytebuff_get_char(p_st_data, &u8_compression);
        if (i_ret != FR_SUCC || u8_compression > 2)
        {
            i_confirm_flg = FR_NO;
            break;
        }
        /* skip CHECKSUM */
        i_ret = bytebuff_skip(p_st_data, 4);
        if (i_ret != FR_SUCC)
        {
            i_confirm_flg = FR_NO;
            break;
        }
        /* get length */
        i_ret = bytebuff_get_int(p_st_data, 4, &u32_payload_len, BYTE_BIG_ENDIAN);
        if (i_ret != FR_SUCC || u32_msg_len != 1 + 1 + 4 + 4 + u32_payload_len)
        {
            i_confirm_flg = FR_NO;
            break;
        }
        i_ret = bytebuff_skip(p_st_data, u32_payload_len);
        if (i_ret != FR_SUCC)
        {
            i_confirm_flg = FR_NO;
            break;
        }
        ui_size_tmp -= u32_msg_len + 4;
    }
    bytebuff_setPosition(p_st_data, ui_position);
    /* 是0.7 */
    if (i_confirm_flg == FR_YES)
    {
        *p_i_msg_ver = KAFKA_VERSION_7;
        return APM_L7_ANALYZE_SUCC;
    }
    return APM_L7_ANALYZE_FAIL;
}

/*******************************************************************************
*  FUNC     :  取得message version
*  ARGS     :  p_st_data(入力buff)
*           :  ui_size(message size)
*           :  p_i_msg_ver(message version)
*           :  p_st_messages(出力message)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_FAIL
*  NOTE     :
*******************************************************************************/
int kafka_parse_messages(Frbyte_Packet *p_st_data
                         , unsigned int ui_size
                         , int i_msg_ver
                         , Frlist_header *p_st_messages)
{
    int i_ret = FR_SUCC;
    uint8_t u8_magic = 0;
    uint8_t u8_compression = 0;
    uint32_t u32_msg_len = 0;
    unsigned int ui_value_len = 0;
    char *p_value = NULL;
    Kafka_message_set_info *p_st_message = NULL;
    Frlist_node *p_st_node = NULL;

    while (ui_size)
    {
        /* get length */
        i_ret = bytebuff_get_int(p_st_data, 4, &u32_msg_len, BYTE_BIG_ENDIAN);
        if (i_ret != FR_SUCC || ui_size < 4 + u32_msg_len)
        {
            frlist_remove_all(p_st_messages, kafka_message_set_node_free);
            return APM_L7_ANALYZE_FAIL;
        }

        /* get MAGIC */
        i_ret = bytebuff_get_char(p_st_data, &u8_magic);
        if (i_ret != FR_SUCC || u8_magic != 0)
        {
            frlist_remove_all(p_st_messages, kafka_message_set_node_free);
            return APM_L7_ANALYZE_FAIL;
        }

        if (i_msg_ver == KAFKA_VERSION_7)
        {
            /* get COMPRESSION */
            i_ret = bytebuff_get_char(p_st_data, &u8_compression);
            if (i_ret != FR_SUCC || u8_compression > 2)
            {
                frlist_remove_all(p_st_messages, kafka_message_set_node_free);
                return APM_L7_ANALYZE_FAIL;
            }
        }

        /* skip CHECKSUM */
        i_ret = bytebuff_skip(p_st_data, 4);
        if (i_ret != FR_SUCC)
        {
            frlist_remove_all(p_st_messages, kafka_message_set_node_free);
            return APM_L7_ANALYZE_FAIL;
        }

        /* get value */
        i_ret = kafka_get_bytes(p_st_data, u8_compression, &p_value, &ui_value_len);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            frlist_remove_all(p_st_messages, kafka_message_set_node_free);
            return i_ret;
        }

        /* 创建message set */
        p_st_message = (Kafka_message_set_info *)frmalloc(sizeof(Kafka_message_set_info));
        if (p_st_message == NULL)
        {
            kafka_free(&p_value);
            frlist_remove_all(p_st_messages, kafka_message_set_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        p_st_message->p_value = p_value;

        /* 创建节点 */
        p_st_node = frlist_node_create(p_st_message);
        if (p_st_node == NULL)
        {
            kafka_message_set_free(&p_st_message);
            frlist_remove_all(p_st_messages, kafka_message_set_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        /* 追加节点 */
        frlist_push(p_st_messages, p_st_node);
        ui_size -= u32_msg_len + 4;
    }

    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  跳过指定长度
*  ARGS     :  p_st_data(入力buff)
*           :  ui_len(长度)
*  RTN      :  APM_L7_ANALYZE_SUCC
*              APM_L7_ANALYZE_FAIL
*  NOTE     :
*******************************************************************************/
int kafka_skip_len(Frbyte_Packet *p_st_data
                  , unsigned int ui_len)
{
    int i_ret = FR_SUCC;

    i_ret = bytebuff_skip(p_st_data, ui_len);
    if (i_ret != FR_SUCC)
    {
        return APM_L7_ANALYZE_FAIL;
    }
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  跳过数组
*  ARGS     :  p_st_data(入力buff)
*           :  ui_len(数组内容长度)
*           :  p_value(数组内容:str int32 [int32 [int8]] str)
*  RTN      :  FR_SUCC
*              FR_FAIL
*  NOTE     :
*******************************************************************************/
int kafka_skip_arr(Frbyte_Packet *p_st_data
                   , unsigned int ui_len
                   , char *p_value)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    int i_arr_count = 0;
    unsigned int ui_len_tmp = 0;
    unsigned int ui_loop = 0;
    char *p_value_tmp = p_value;
    unsigned int ui_left = 0;

    /* get array count */
    i_ret = kafka_get_int(p_st_data, &i_arr_count);
    if (i_ret != APM_L7_ANALYZE_SUCC || i_arr_count < 0)
    {
        return APM_L7_ANALYZE_FAIL;
    }

    while (i_arr_count--)
    {
        while (ui_loop < ui_len)
        {
            switch (p_value[ui_loop])
            {
                case '[':
                    ui_left = 1;
                    p_value_tmp += ui_loop + 1;
                    ui_loop += 1;
                    while (ui_loop < ui_len)
                    {
                        if (p_value[ui_loop] == '[')
                        {
                            ui_left += 1;
                        }
                        else if (p_value[ui_loop] == ']')
                        {
                            ui_left -= 1;
                            if (ui_left == 0)
                            {
                                break;
                            }
                        }
                        ui_loop++;
                        ui_len_tmp++;
                    }
                    /* []不匹配 */
                    if (ui_left)
                    {
                        return APM_L7_ANALYZE_FAIL;
                    }
                    i_ret = kafka_skip_arr(p_st_data, ui_len_tmp, p_value_tmp);
                    break;

                case KAFKA_CHAR:
                    i_ret = kafka_skip_len(p_st_data, 1);
                    break;

                case KAFKA_SHORT:
                    i_ret = kafka_skip_len(p_st_data, 2);
                    break;

                case KAFKA_INT:
                    i_ret = kafka_skip_len(p_st_data, 4);
                    break;

                case KAFKA_LONG:
                    i_ret = kafka_skip_len(p_st_data, 8);
                    break;

                case KAFKA_STRINT:
                    i_ret = kafka_skip_string(p_st_data);
                    break;

                case KAFKA_BYTES:
                    i_ret = kafka_skip_bytes(p_st_data);
                    break;

                default:
                    return APM_L7_ANALYZE_FAIL;
            }

            if (i_ret != APM_L7_ANALYZE_SUCC)
            {
                return APM_L7_ANALYZE_FAIL;
            }
            ui_loop++;
        }
        ui_loop = 0;
    }

    return APM_L7_ANALYZE_SUCC;
}

